package com.whz.disruptor.configuration;

import com.lmax.disruptor.dsl.Disruptor;
import com.whz.disruptor.event.MessageEventConsumer;
import com.whz.disruptor.handler.ConsumerHandler;
import com.whz.disruptor.properties.DisruptorProperties;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.CollectionUtils;
import com.whz.disruptor.broker.DisruptorBrokerBean;
import com.whz.disruptor.handler.ProductHandler;

import java.util.Map;

/**
 * @author whz
 */
@Slf4j
@Configuration
@ConditionalOnClass({ Disruptor.class })
@EnableConfigurationProperties(DisruptorProperties.class)
@ConditionalOnProperty(name = "tongyu.starter.disruptor.enable", havingValue = "true", matchIfMissing = true)
public class DisruptorAutoConfiguration implements ApplicationContextAware {

    private ApplicationContext applicationContext;

    @Bean
    public DisruptorBrokerBean createDisruptorBroker(DisruptorProperties disruptorProperties) {
        Map<String, MessageEventConsumer> map = applicationContext.getBeansOfType(MessageEventConsumer.class);
        if (CollectionUtils.isEmpty(map)) {
            log.error("Not Found Beans of the Type[MessageEventConsumer], the default one started.");
            return new DisruptorBrokerBean(disruptorProperties, ConsumerHandler.DEFAULT_CONSUMER_HANDLER);
        }
        return new DisruptorBrokerBean(disruptorProperties,
                                       map.values()
                                          .stream()
                                          .map(ConsumerHandler::new)
                                          .toArray(ConsumerHandler[]::new));
    }

    @Bean
    public ProductHandler createProductHandler(DisruptorBrokerBean disruptorBrokerBean) {
        return new ProductHandler(disruptorBrokerBean.getRingBuffer());
    }


    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}